using System;
using System.Collections.Generic;
using System.Threading;
using NLog;

namespace CoreLibrary.Concurrent {

    public class AsyncBlockingTaskQueue : IDisposable {

        public enum Priority {
            High = 1,
            Normal = 0
        }

        public delegate void ExecutorDelegate(object task, CompletionCallback completionCallback);
        public delegate void CompletionCallback();
        public delegate void ErrorHandlerDelegate(object task, Exception error);

        public ExecutorDelegate     Executor { get; set; }
        public ErrorHandlerDelegate ErrorHandler { get; set; }
        public int MaxThread { get; private set; }
        private readonly object _locker = new object();
        private readonly LinkedList<object> _queue   = new LinkedList<object>();
        private readonly HashSet<object> _queueIndex = new HashSet<object>();
        private readonly Semaphore _semaphore;
        private readonly ManualResetEvent _starter = new ManualResetEvent(false);

        private int _activeCount;

        public AsyncBlockingTaskQueue(int maxThread) {
            MaxThread = maxThread;
            _semaphore = new Semaphore(maxThread, maxThread);
            ThreadPool.QueueUserWorkItem(Dispatch, null);
        }

        public void Start() {
            _starter.Set();
        }

        public void Stop() {
            _starter.Reset();
        }

        public int Count() {
            return CountInactive() + _activeCount;
        }

        public int CountActive() {
            return _activeCount;
        }

        public int CountInactive() {
            lock (_locker) {
                return _queue.Count;
            }
        }

        public List<object> Get(int offset, int max) {
            Asserts.True(offset >= 0, "offset must be zero or a positive number");
            Asserts.True(max > 0 && max <= 100, "max must be between 1 and 100");

            lock (_locker) {
                var result = new List<object>();
                if (offset >= _queue.Count) {
                    return result;
                }

                LinkedList<object>.Enumerator enumerator = _queue.GetEnumerator();
                while (offset > 0 && enumerator.MoveNext()) {
                    --offset;
                }

                while (max > 0 && enumerator.MoveNext()) {
                    result.Add(enumerator.Current);
                    --max;
                }
                return result;
            }
        }

        /// <summary>
        /// Check whether an object is in the queue. This is an expensive operation when the queue is large.
        /// </summary>
        /// <param name="data"></param>
        /// <returns></returns>
        public bool Has(object data) {
            return _queueIndex.Contains(data);
        }

        public void Enqueue(object data, Priority priority) {
            if (data == null) {
                throw new Exception("Data must not be null.");
            }
            if (Executor == null) {
                throw new Exception("Executor has not been set.");
            }
            InternalEnqueueTask(data, priority);
        }

        public bool EnqueueIfAbsent(object data, Priority priority) {
            if (data == null) {
                throw new Exception("Data must not be null.");
            }
            if (Executor == null) {
                throw new Exception("Executor has not been set.");
            }
            lock (_locker) {
                var absent = !_queueIndex.Contains(data);
                if (absent) {
                    InternalEnqueueTask(data, priority);
                }
                return absent;
            }
        }

        public bool Remove(object data) {
            if (data == null) {
                throw new Exception("Data must not be null.");
            }
            lock (_locker) {
                if (_queueIndex.Remove(data)) {
                    _queue.Remove(data);
                    return true;
                }
                return false;
            }
        }

        public void Dispose() {
            // Enqueue a null makes dispatch thread exit.
            InternalEnqueueTask(null, Priority.High);
        }

        private void InternalEnqueueTask(object data, Priority priority) {
            lock (_locker) {
                if (priority == Priority.High) {
                    _queue.AddFirst(data);
                } else {
                    _queue.AddLast(data);
                }
                if (data != null) {
                    _queueIndex.Add(data);
                }
                Monitor.Pulse(_locker);
            }
        }

        private void Dispatch(object state) {
            while (true) {
                _starter.WaitOne();
                object task;
                lock (_locker) {
                    while (_queue.Count == 0) {
                        Monitor.Wait(_locker);
                    }
                    task = _queue.First.Value;
                    _queue.RemoveFirst();
                    if (task != null) {
                        _queueIndex.Remove(task);
                    }
                }

                if (task == null) {
                    return; // This signals our exit
                }
                _semaphore.WaitOne();
                Interlocked.Increment(ref _activeCount);
                ThreadPool.QueueUserWorkItem(Process, task);
            }
        }

        private void Process(object task) {
            int i = 0;
            try {
                Executor(task, () => {
                    if (Interlocked.CompareExchange(ref i, 1, 0) == 0) {
                        _semaphore.Release();
                        Interlocked.Decrement(ref _activeCount);
                    }
                });
            } catch (Exception e) {
                if (Interlocked.CompareExchange(ref i, 2, 0) == 0) {
                    _semaphore.Release();
                    Interlocked.Decrement(ref _activeCount);
                }
                if (ErrorHandler != null) {
                    try {
                        ErrorHandler(task, e);
                    } catch (Exception e2) {
                        LogManager.GetCurrentClassLogger().ErrorException("Error handling failed: " + e.Message, e2);
                    }
                }
            }
        }

    }
}